package com.gzx.qc.modular.text.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.gzx.qc.modular.text.entity.GzxAsrVoiceText;
import com.gzx.qc.modular.text.mapper.GzxAsrVoiceTextMapper;
import com.gzx.qc.modular.text.service.GzxAsrVoiceTextService;
import com.gzx.qc.modular.text.vo.GzxAsrVoiceTextVo;
import com.gzx.qc.modular.voice.check.entity.GzxVoiceCheckDetail;
import com.gzx.qc.modular.voice.check.service.GzxVoiceCheckService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author xbl
 */
@Slf4j
@Service
@DS("qc")
public class GzxAsrVoiceTextServiceImpl implements GzxAsrVoiceTextService {

    private final GzxAsrVoiceTextMapper gzxAsrVoiceTextMapper;

    private final GzxVoiceCheckService voiceCheckService;

    @Autowired
    public GzxAsrVoiceTextServiceImpl(GzxAsrVoiceTextMapper gzxAsrVoiceTextMapper, GzxVoiceCheckService voiceCheckService) {
        this.gzxAsrVoiceTextMapper = gzxAsrVoiceTextMapper;
        this.voiceCheckService = voiceCheckService;
    }

    //每次获取上次处理之后的最大id
    @DS("qc")
    @Override
    public int getMaxVoiceId() {
        return gzxAsrVoiceTextMapper.getMaxVoiceId();
    }

    @DS("qc")
    @Override
    public void batchInsert(List<GzxAsrVoiceTextVo> textList) {
        gzxAsrVoiceTextMapper.batchInsert(textList);
    }

    //Kafka消费
    @Override
    public void kafkaConsumer(List<ConsumerRecord<String, String>> recordList) {
        List<GzxAsrVoiceTextVo> voiceTextList = new ArrayList<>();
        //接到 Kafka 的 asr-text-topic 消息
        log.info("从MQ中批量获取asr：本批次处理event数量{}", recordList.size());
        for (ConsumerRecord<String, String> consumerRecord : recordList) {
            log.info("Kafka收到asr消息：consumerRecord -> {}", consumerRecord.value());
            GzxAsrVoiceTextVo text = JSONObject.parseObject(consumerRecord.value(), GzxAsrVoiceTextVo.class);
            voiceTextList.add(text);
        }
        //遍历 voiceTextList ，说话内容content为空的都过滤掉
        voiceTextList = voiceTextList.stream()
                .filter(vt -> vt.getContent() != null && !vt.getContent().isEmpty())
                .collect(Collectors.toList());

        if (!voiceTextList.isEmpty()) {
            log.info("本批次入库asr数量{}", voiceTextList.size());
            batchInsert(voiceTextList);
        }
    }

    /**
     * 根据CallID获取asr通话文本
     */
    @DS("qc")
    @Override
    public List<GzxAsrVoiceTextVo> getTextByCallId(String callId) {
        List<GzxAsrVoiceTextVo> asrVoiceTexts = gzxAsrVoiceTextMapper.selectVoList(Wrappers.<GzxAsrVoiceText>lambdaQuery()
                .eq(GzxAsrVoiceText::getCallId, callId)
                .orderByAsc(GzxAsrVoiceText::getChatTime));
        LambdaQueryWrapper<GzxVoiceCheckDetail> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(GzxVoiceCheckDetail::getCallId, callId);
        //校验是不是命中关键字了
        List<GzxVoiceCheckDetail> gzxVoiceCheckDetails = voiceCheckService.getGzxVoiceCheckDetailBywarpper(wrapper);
        Map<String, GzxAsrVoiceTextVo> textAndGzxHtVoiceTextHashMap = asrVoiceTexts.stream()
                .collect(Collectors.toMap(GzxAsrVoiceTextVo::getContent, v -> v, (existing, replacement) -> existing));
        for (GzxVoiceCheckDetail gzxVoiceCheckDetail : gzxVoiceCheckDetails) {
            if (textAndGzxHtVoiceTextHashMap.containsKey(gzxVoiceCheckDetail.getContent())) {
                String keyword = gzxVoiceCheckDetail.getKeyword();
                String replace = textAndGzxHtVoiceTextHashMap.get(gzxVoiceCheckDetail.getContent()).getContent();
                //给它一个高亮的样式
                for (String s : keyword.split(",")) {
                    String replacement = "<span class=\"highlight\">" + s + "</span>";
                    replace = replace.replace(s, replacement);
                }
                //命中的关键词set到content中
                textAndGzxHtVoiceTextHashMap.get(gzxVoiceCheckDetail.getContent()).setContent(replace);
            }
        }
        return asrVoiceTexts;
    }


}
